package com.nageoffer.jobtrain.projects.mq.consumer;

import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.util.ObjectUtil;
import com.nageoffer.jobtrain.projects.common.constant.RocketMQContent;
import com.nageoffer.jobtrain.projects.dao.entity.OjQuestionESDO;
import com.nageoffer.jobtrain.projects.elasticsearch.OjQuestionRepository;
import com.nageoffer.jobtrain.projects.mq.event.CanalBinlogEvent;
import lombok.RequiredArgsConstructor;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.stereotype.Component;

import java.text.SimpleDateFormat;
import java.util.Map;

/**
 * 通过 Canal 监听管理员操作Oj题目表 Binlog 投递消息队列消费
 */
@Slf4j
@Component
@RequiredArgsConstructor
@RocketMQMessageListener(
        topic = RocketMQContent.CANAL_OJ_QUESTION_TOPIC_KEY,
        consumerGroup = RocketMQContent.CANAL_OJ_QUESTION_CG_KEY
)
public class CanalBinlogSyncOjQuestionConsumer implements RocketMQListener<CanalBinlogEvent> {
    private final OjQuestionRepository ojQuestionRepository;

    @SneakyThrows
    @Override
    public void onMessage(CanalBinlogEvent canalBinlogEvent) {
        // 获取到操作之后的Oj题目数据
        Map<String, Object> data = CollUtil.getFirst(canalBinlogEvent.getData());
        // 定义日期格式
        SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
        if (ObjectUtil.equal(canalBinlogEvent.getType(), "INSERT") || ObjectUtil.equal(canalBinlogEvent.getType(), "UPDATE")) {
            OjQuestionESDO ojQuestionESDO = OjQuestionESDO.builder()
                    .id(Long.valueOf(data.get("id").toString()))
                    .title(data.get("title").toString())
                    .difficulty(Integer.valueOf(data.get("difficulty").toString()))
                    .timeLimit(Integer.valueOf(data.get("time_limit").toString()))
                    .spaceLimit(Integer.valueOf(data.get("space_limit").toString()))
                    .content(data.get("content").toString())
                    .defaultCode(data.get("default_code").toString())
                    .questionCase(data.get("question_case").toString())
                    .mainFunction(data.get("main_function").toString())
                    .createTime(sdf.parse(data.get("create_time").toString()))
                    .build();
            // 此时是新增OJ题目，根据ID进行判断，如果存在就覆盖，不存在直接插入到ES中
            ojQuestionRepository.save(ojQuestionESDO);
        } else if (ObjectUtil.equal(canalBinlogEvent.getType(), "DELETE")) {
            // 根据ID进行删除ES中的元素
            ojQuestionRepository.deleteById(Long.valueOf(data.get("id").toString()));
        }
    }
}